package com.etlmaster.executor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

import com.etlmaster.executor.steps.control.RingStepMeta;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class Job implements Cloneable{
	
	private JobMeta jobMeta;
	
	private Map<String,StepMeta> stepMetaIndexs = new HashMap<String,StepMeta>();
	
	private Map<String,List<String>> preStepIndex = new HashMap<String,List<String>>();
	
	private Map<String,List<String>> nextStepIndex = new HashMap<String,List<String>>();//完成时
	
	//需要下列步骤来控制事务
	private Map<String,List<String>> successStepIndex=new HashMap<String,List<String>>();//成功时
	
	private Map<String,List<String>> failedStepIndex=new HashMap<String,List<String>>();//失败时
	
	private Map<String,List<String>> completeStepIndex = new HashMap<String,List<String>>();//完成时
	
	private Map<String,StepMeta> runOffStepMetas = new LinkedHashMap<String,StepMeta>();

	private boolean isTmpJob=false;
	public AtomicBoolean successed = new AtomicBoolean(false);
	
	public Job(JobMeta jobMeta) {
		super();
		this.jobMeta = jobMeta;
	}

	
	public StepMeta getRunOffStepMetaById(String subsession_id,String id){
		return runOffStepMetas.get(subsession_id+"_"+id);
	}
	public JobMeta getMeta() {
		return jobMeta;
	}

	public void setMeta(JobMeta jobMeta) {
		this.jobMeta = jobMeta;
	}
	
	public Map<String, List<String>> getCompleteStepIndex() {
		return completeStepIndex;
	}

	public Map<String, List<String>> getPreStepIndex() {
		return preStepIndex;
	}

	public Map<String, StepMeta> getStepMetaIndexs() {
		return stepMetaIndexs;
	}

	public Map<String,List<String>> getSuccessStepIndex() {
		return successStepIndex;
	}

	public Map<String,List<String>> getFailedStepIndex() {
		return failedStepIndex;
	}

	public Map<String,List<String>> getNextStepIndex() {
		return nextStepIndex;
	}
	
	public List<StepMeta> getRootStepMetas(){
		List<StepMeta> result = new ArrayList<StepMeta>();
		for (Iterator iterator = stepMetaIndexs.entrySet().iterator(); iterator.hasNext();) {
			Map.Entry<String, StepMeta> entry = (Map.Entry<String, StepMeta>) iterator.next();
			if(entry.getValue() instanceof RingStepMeta ||
				!preStepIndex.containsKey(entry.getKey()) || 
					(preStepIndex.get(entry.getKey())!=null && preStepIndex.get(entry.getKey()).size()==0 )){
				result.add(entry.getValue());
			}
		}
		return result;
	}
	
	private List<StepMeta> getChildrenStepMeta(String stepId,Map<String,List<String>> parentIndex){
		List<StepMeta> result = new ArrayList<StepMeta>();
		List<String> stepIds = parentIndex.get(stepId);
		if(stepIds !=null){
			for (String sid : stepIds) {
				result.add(stepMetaIndexs.get(sid));
			}
		}
		return result;
	}
	
	public List<StepMeta> getPreStepMetas(String stepId){
		return getChildrenStepMeta(stepId, preStepIndex);
	}
	
	public List<StepMeta> getNextStepMetas(String stepId){
		return getChildrenStepMeta(stepId, nextStepIndex);
	}
	
	public List<StepMeta> getCompleteStepMetas(String stepId){
		return getChildrenStepMeta(stepId, completeStepIndex);
	}
	
	public List<StepMeta> getSucessStepMetas(String stepId){
		return getChildrenStepMeta(stepId, successStepIndex);
	}
	
	public List<StepMeta> getFailStepMetas(String stepId){
		return getChildrenStepMeta(stepId, failedStepIndex);
	}
	
	public List<StepMeta> getNextExecStepMetas(String stepId,StepState stepState) throws EtlMasterException{
		
		int stepStateCode = stepState.getStatus();
		List<StepMeta> nextSteps = null;
		
		//fengwen 20150325 修改陈成功、失败、完成时执行判断方法
		if(StepState.RUNNING_SUCCESSED == stepStateCode){
			//step.setSuccessed(true);
			if(!"-1".equals(getSucessStepId(stepId)) && !"99".equals(getSucessStepId(stepId)) && getSucessStepId(stepId)!=null){
				
				nextSteps = getSucessStepMetas(stepId);
			}else if( "-1".equals(getFailStepId(stepId)) ){
				nextSteps = getNextStepMetas(stepId);
			}
		}else{//步骤没有执行成功，直接通知Executor汇报错误并中止程序
			String completeStep=getCompleteStepId(stepId);
			if (!"-1".equals(completeStep) && completeStep!=null) {
				//继续执行
				//需要值挡墙状态为成功么？setSuccessed(true);
				nextSteps=getCompleteStepMetas(stepId);
				
			}
			if(!"-1".equals(getFailStepId(stepId)) && getFailStepId(stepId)!=null){
				//step.setSuccessed(true);
				nextSteps=getFailStepMetas(stepId);
			}
			
			if(nextSteps==null || nextSteps.size()==0){
				throw new EtlMasterException("步骤运行失败"+stepStateCode+"，"+stepId+"且没有下一个步骤!");
			}
		}
		return nextSteps;
	}
	
	public String getCompleteStepId(String stepId){
		List<String> stepIds = completeStepIndex.get(stepId);
		String ids="";
		if(stepIds !=null){
			for (String sid : stepIds) {
				ids=ids+sid;
			}
		}
		return ids;
	}
	
	public String getSucessStepId(String stepId){
		List<String> stepIds = successStepIndex.get(stepId);
		String ids="";
		if(stepIds !=null){
			for (String sid : stepIds) {
				ids=ids+sid;
			}
		}
		return ids;
	}

	public String getFailStepId(String stepId){
		List<String> stepIds = failedStepIndex.get(stepId);
		String ids="";
		if(stepIds !=null){
			for (String sid : stepIds) {
				ids=ids+sid;
			}
		}
		return ids;
	}

	public boolean getSuccessed() {
		return successed.get();
	}

	public void setSuccessed(boolean successed) {
		this.successed.set(successed);
	}
	
	public void addVariables(String uuid,Map<String,Object> attachVars){
		JobPool.getVariable(jobMeta.getSeesionId()).putAll(attachVars);
	}
	
	public void putVariable(String key, Object value){
		JobPool.getVariable(jobMeta.getSeesionId()).put(key,value);
	}
	
	public Object getVariable(String key){
		return JobPool.getVariable(jobMeta.getSeesionId()).get(key);
	}

	public Map<String, Object> getVariables() {
		return JobPool.getVariable(jobMeta.getSeesionId());
	}
	
	public String getSeesionId() {
		return jobMeta.getSeesionId();
	}
	protected Object clone() throws CloneNotSupportedException {

		Job jobInstance = (Job) super.clone();
		jobInstance.jobMeta=(JobMeta) jobInstance.jobMeta.clone();
		String uuid = UUID.randomUUID().toString();
		jobInstance.jobMeta.setSeesionId(uuid);
		Map<String,StepMeta> newSteps=new HashMap<String,StepMeta>();
		Iterator keySet=this.getStepMetaIndexs().keySet().iterator();
		while(keySet.hasNext()){
			String key=(String) keySet.next();
			StepMeta meta=(StepMeta) this.getStepMetaIndexs().get(key).clone();
			StepState state=new StepState(uuid,meta.getId());
			meta.setStepState(state);
			meta.setSession_id(uuid);
			newSteps.put(key,meta);
		}
		jobInstance.stepMetaIndexs=newSteps;
		jobInstance.runOffStepMetas=new LinkedHashMap<String,StepMeta>();

		return jobInstance;

	}

	public void putRunOffStepMetas(String sub_session_id,String id,StepMeta meta){
		Iterator iter = runOffStepMetas.keySet().iterator();
		List<String> keys=new ArrayList<String>();
		while (iter.hasNext()) {
			String key = (String) iter.next();
			if(key.split("_")[1].equals(id)){
				keys.add(key);
			}
		}
		for(String key:keys){
			runOffStepMetas.remove(key);
		}
		runOffStepMetas.put(sub_session_id+"_"+id, meta);
	}


	public boolean isTmpJob() {
		return isTmpJob;
	}


	public void setTmpJob(boolean isTmpJob) {
		this.isTmpJob = isTmpJob;
	}

}
